#include "pqxx/pqxx"
using namespace pqxx;

#include "pqxx_connection.hpp"

namespace topdbs
{
	pqxx_connection::pqxx_connection()
		: m_pConnection(nullptr)
		, m_isTransaction(false)
		, m_isConnected(false)
		, m_port(0)
		, m_work(nullptr)
	{
	}

	pqxx_connection::~pqxx_connection()
	{
		disConnect();
	}
	
	//! Opens the connection to a data source.
	int  pqxx_connection::connect(
		eDBSClient_t dbsClientType,
		const string& dbString,
		const string& user,
		const string& pass,
		void* pOption  )
	{
		m_user = user;
		m_pass = pass;
		m_dbsClient = dbsClientType;

		if(dbString.empty())
		{
			m_sqlerr = "The database connection string is empty.";
			return ERR_DBS_FAIL;
		}
		topdbs_dbstring_parse(dbString,m_dbname,m_host,m_port);
		dbslog(DBS_DEBUG,"dbString=%s user=%s pass=***",dbString.c_str(),m_user.c_str());

		if(m_host.empty() || m_port == 0)
		{
			m_sqlerr = "The database connection string (hostname or port) is invalid.";
			return ERR_DBS_FAIL;
		}
		disConnect();
		m_pConnection = connection_create();
		return m_pConnection ? ERR_DBS_OK : ERR_DBS_FAIL;
	}

	//! Returns the connection state for a particular connection object.	
	bool pqxx_connection::isConnected() 
	{
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection,false);	
		return m_pConnection->is_open();;
	}

	int  pqxx_connection::reConnect() 
	{
		disConnect();
		m_pConnection = connection_create();
		return m_pConnection ? ERR_DBS_OK : ERR_DBS_FAIL;;
	}

	//! Disconnects the connection from the database.
	int  pqxx_connection::disConnect() 
	{
		if(nullptr != m_pConnection)
		{
			if(isTransaction())
			{
				rollback();
			}		
			if(m_work)
			{
				delete m_work;
				m_work = nullptr;
			}		
			m_pConnection = nullptr;
			m_isConnected = false;
			m_isTransaction = false;
		}	
		return ERR_DBS_OK;
	}
		
	//! Begin transactions.
	int  pqxx_connection::begin() 
	{
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection,ERR_DBS_UNKOWN);
		if(isTransaction())
		{
			return ERR_DBS_OK;
		}

		try
		{
			m_work = new work(*m_pConnection);
		}
		catch(const pqxx::pqxx_exception& e)
		{
			m_sqlerr = e.base().what();
			dbslog(DBS_ERROR,"begin() TransactionException  erortext[%s]",m_sqlerr.c_str());
			return ERR_DBS_FAIL;		
		}
		lockwait();
		m_isTransaction = true;
		return ERR_DBS_OK;
	}

	//! Commits transactions.
	int  pqxx_connection::commit() 
	{
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection,ERR_DBS_UNKOWN);
		if(!isTransaction())
		{
			return ERR_DBS_OK;
		}
		try
		{	
			m_work->commit();
		}
		catch(const pqxx::pqxx_exception& e)
		{
			m_sqlerr = e.base().what();
			dbslog(DBS_ERROR,"commit() TransactionException  erortext[%s]",m_sqlerr.c_str());
			return ERR_DBS_EXCEPTION;
		}	
		m_isTransaction = false;
		return ERR_DBS_OK;
	}

	//! Rollback transactions.
	int  pqxx_connection::rollback() 
	{
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection,ERR_DBS_UNKOWN);
		if(!isTransaction())
		{
			return ERR_DBS_OK;
		}	
		try
		{	
			m_work->abort();
		}
		catch(const pqxx::pqxx_exception& e)
		{
			m_sqlerr = e.base().what();
			dbslog(DBS_ERROR,"rollback() TransactionException erortext[%s]",m_sqlerr.c_str());
			return ERR_DBS_EXCEPTION;
		}
		m_isTransaction = false;
		return ERR_DBS_OK;
	}

	//! Returns transactions.
	bool pqxx_connection::isTransaction() 
	{
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection,false);	
		return m_isTransaction;
	}

	//! Sets the connection to be committed automatically.
	int  pqxx_connection::autoCommit(bool on) 
	{
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection,ERR_DBS_UNKOWN);
		//m_pConnection->set_variable("AUTOCOMMIT",isAutoFlag?"ON":"OFF");
		return ERR_DBS_OK;
	}

	//! Returns the native connection
	void*  pqxx_connection::getConnection() 
	{
		return m_pConnection;
	}

	//! Returns the current this
	void* pqxx_connection::from_this() 
	{
		return this;
	}

	int pqxx_connection::lockwait()
	{	
		char val[256] = {0};
		CHK_SESSION_HANDLE_ERROR_RETURN(m_pConnection,ERR_DBS_UNKOWN);
		if(m_locktime < 1 )
		{
			char* psEnvLockTime = getenv("DB_LOCK_WAIT_TIME");
			if(psEnvLockTime)
			{
				m_locktime = atoi(psEnvLockTime); 
			}		
		}
		
		if(m_locktime == 0 || m_locktime > 900 ) 
		{
			m_locktime = TOPDBS_DEFAULT_LOCK_WAIT_TIME;
		}			

		try
		{	
			sprintf(val,"%d",m_locktime);
#if PQXX_VERSION_MAJOR > 6
				
#else	
			m_pConnection->set_variable("lock_timeout",val);
#endif			
		}
		catch(const pqxx::pqxx_exception& ex)
		{
			m_sqlerr = ex.base().what();
			dbslog(DBS_ERROR,"lockwait Exception erortext[%s]",m_sqlerr.c_str());		
			return ERR_DBS_FAIL;
		}
		return ERR_DBS_OK;
	}

	pqxx::connection* pqxx_connection::connection_create( )
	{
		string dbConnStr;
		pqxx::connection* session_ptr = nullptr;
		try {
			
			topdbs_string::format(dbConnStr,"dbname = %s user = %s password=%s hostaddr=%s port=%d",
				m_dbname.c_str(),m_user.c_str(),m_pass.c_str(),m_host.c_str(),m_port);

			session_ptr = new pqxx::connection(dbConnStr); 
			if(m_charset.empty()) m_charset = "GBK";
			session_ptr->set_client_encoding(m_charset);
		}
		catch(const pqxx::pqxx_exception& ex)
		{
			m_sqlerr = ex.base().what();
			dbslog(DBS_ERROR,"connection_create() ConnectionFailedException %s !!",m_sqlerr.c_str());
			return session_ptr;
		}
		return session_ptr;		
	}
}